package com.mini.rpc.codec;

import com.mini.rpc.common.MiniRpcRequest;
import com.mini.rpc.common.MiniRpcResponse;
import com.mini.rpc.protocol.MiniRpcProtocol;
import com.mini.rpc.protocol.MsgHeader;
import com.mini.rpc.protocol.MsgType;
import com.mini.rpc.protocol.ProtocolConstants;
import com.mini.rpc.serialization.RpcSerialization;
import com.mini.rpc.serialization.SerializationFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

/**
 * 解码器
 * <p> 将字节流数据解码为消息对象，并传递给下一个 Inbound 处理器。
 * <ul>
 *     <li>在RPC请求调用场景中，服务提供者使用该编码器，将二进制数据解码成
 *     {@link MiniRpcProtocol<MiniRpcRequest>} 对象之后，再传递给
 *     {@link com.mini.rpc.handler.RpcRequestHandler} 执行RPC请求调用</li>
 *     <li>在RPC响应场景中，服务消费者使用该编码器，将二进制数据解码成
 *     {@link MiniRpcProtocol<MiniRpcResponse>} 对象之后，再传递给
 *     {@link com.mini.rpc.handler.RpcResponseHandler} 执行RPC请求响应</li>
 * </ul>
 *
 */
public class MiniRpcDecoder extends ByteToMessageDecoder {

    /* 解码 <br>
    +---------------------------------------------------------------+
    | 魔数 2byte | 协议版本号 1byte | 序列化算法 1byte | 报文类型 1byte  |
    +---------------------------------------------------------------+
    | 状态 1byte |        消息 ID 8byte     |      数据长度 4byte     |
    +---------------------------------------------------------------+
    |                   数据内容 （长度不定）                          |
    +---------------------------------------------------------------+
    */
    @Override
    public final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 只有当 ByteBuf 中内容大于协议头 Header 的固定的 18 字节时，才开始读取数据。
        if (in.readableBytes() < ProtocolConstants.HEADER_TOTAL_LEN) {
            return;
        }
        // 标记读指针位置
        in.markReaderIndex();

        short magic = in.readShort();
        if (magic != ProtocolConstants.MAGIC) {
            throw new IllegalArgumentException("magic number is illegal, " + magic);
        }

        byte version = in.readByte();
        byte serializeType = in.readByte();
        byte msgType = in.readByte();
        byte status = in.readByte();
        long requestId = in.readLong();

        int dataLength = in.readInt();
        if (in.readableBytes() < dataLength) {
            // 当 ByteBuf 中可读字节长度小于协议体 Body 的长度时
            // 使用 resetReaderIndex() 还原读指针位置
            // 说明现在 ByteBuf 中可读字节还不够一个完整的数据包
            in.resetReaderIndex();
            return;
        }
        byte[] data = new byte[dataLength];
        in.readBytes(data);

        MsgType msgTypeEnum = MsgType.findByType(msgType);
        if (msgTypeEnum == null) {
            return;
        }

        MsgHeader header = new MsgHeader();
        header.setMagic(magic);
        header.setVersion(version);
        header.setSerialization(serializeType);
        header.setStatus(status);
        header.setRequestId(requestId);
        header.setMsgType(msgType);
        header.setMsgLen(dataLength);

        RpcSerialization rpcSerialization = SerializationFactory.getRpcSerialization(serializeType);
        // 根据不同的报文类型 MsgType，反序列化出不同的协议体对象。
        switch (msgTypeEnum) {
            case REQUEST:
                // 服务提供者需要将协议体内容反序列化成 MiniRpcRequest 对象
                MiniRpcRequest request = rpcSerialization.deserialize(data, MiniRpcRequest.class);
                if (request != null) {
                    MiniRpcProtocol<MiniRpcRequest> protocol = new MiniRpcProtocol<>();
                    protocol.setHeader(header);
                    protocol.setBody(request);
                    out.add(protocol);
                }
                break;
            case RESPONSE:
                // 在 RPC 结果响应的场景下，服务消费者需要将协议体内容反序列化成 MiniRpcResponse 对象。
                MiniRpcResponse response = rpcSerialization.deserialize(data, MiniRpcResponse.class);
                if (response != null) {
                    MiniRpcProtocol<MiniRpcResponse> protocol = new MiniRpcProtocol<>();
                    protocol.setHeader(header);
                    protocol.setBody(response);
                    out.add(protocol);
                }
                break;
            case HEARTBEAT:
                // TODO
                break;
        }
    }
}
